package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo8GroupBy {
  def main(args: Array[String]): Unit = {

    // 统计班级人数
    // 构建Spark上下文环境
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo8GroupBy")
    conf.setMaster("local")
    val sc: SparkContext = new SparkContext(conf)


    // 1、读取students数据
    val linesRDD: RDD[String] = sc.textFile("Spark/data/stu/students.txt")

    // 2、将数据构建成 (班级,1)

    val clazzRDD: RDD[(String, Int)] = linesRDD.map(line => (line.split(",")(4), 1))

    // 3、按班级分组
    /**
     * groupBy：转换算子
     * 需要指定按什么进行排名
     */
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = clazzRDD.groupBy(kv => kv._1)

    groupRDD.foreach(println)
    // 4、统计班级人数
    groupRDD.map {
      case (clazz: String, clazzIter: Iterable[(String, Int)]) => {
        clazz + "," + clazzIter.map(_._2).sum
      }
    }.foreach(println)

    /**
     * groupByKey:转换算子
     * 分区类算子只能作用在K-V格式的RDD上
     * 在当前程序中，linesRDD不是K-V格式的，所以没有groupByKey算子
     *
     * groupByKey算子会默认按照Key进行分组，结果同groupBy类似，但有细微差异
     * 这两个group算子都会返回K-V格式的
     * K：指定的分组字段（groupBy）、K-V格式的RDD的Key（groupByKey）
     * V：符合相同分组条件的一个整体（groupBy），只会返回Value（groupByKey）
     *
     * (理科二班,CompactBuffer((理科二班,1), (理科二班,1),......) groupBygroupByKey的结果
     * (理科二班,CompactBuffer(1, 1, 1, 1) groupByKey的结果
     */


    clazzRDD.groupByKey().map(kv => (kv._1, kv._2.sum)).foreach(println)


  }

}
